package com.atguigu.edu.realtime220815.app.dwd.db;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime220815.util.KafkaUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.functions.TimedOutPartialMatchHandler;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * @Classname DwsTrafficChIsNewUserSkipWindow dws流量域 各渠道 各类别 用户跳出窗口汇总表
 * @Description TODO
 * @Date 2023/2/17 14:03
 * @Created by lzx
 */
public class DwdTrafficUserJumpDetail {
    public static void main(String[] args) throws Exception {
        /**
         * 主要任务 : 过滤用户跳出明细数据
         * 跳出是指会话中只有一个页面的访问行为,只要筛选页面数为 1 的会话即可获取跳出明细数据。
         *
         * 如果能判定首页日志之后没有同一会话的页面访问记录同样可以筛选跳出数据。如果日志数据完全有序,会话页面不存在交叉情况,
         * 则跳出页面的判定可以分为三种情况
         * ①两条紧邻的首页日志进入算子,可以判定第一条首页日志所属会话为跳出会话;
         * ②第一条首页日志进入算子后,接收到的第二条日志为非首页日志,则第一条日志所属会话不是跳出会话;
         * ③第一条首页日志进入算子后,没有收到第二条日志,此时无法得出结论,必须继续等待。但是无休止地等待显然是不现实的。
         * 因此,人为设定超时时间,超时时间内没有第二条数据就判定为跳出行为,这是一种近似处理,存在误差,但若能结合业务场景设置合理
         * 的超时时间,误差是可以接受的。本程序为了便于测试,设置超时时间为 10s,为了更快看到效果可以设置更小的超时时间,生产环境的
         * 设置结合业务需求确定。
         * 由上述分析可知,情况 ① 的首页数据和情况 ③ 中的超时数据为跳出明细数据。
         *
         * 主要步骤:
         * TODO 1.创建流式执行环境
         * TODO 2.检查点设置
         * TODO 3.从kafka的dwd_traffic_page_log中读取数据,转化格式 jsonStr --> jsonObj , 设置水位线,分组
         * TODO 4.开始flink cep的三步
         *      4.1 设置模式
         *      4.2 使用CEP将流与模式结合在一起
         *      4.3 将符合模式的数据过滤出来进行进一步的处理
         */
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);


        /*env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.setStateBackend(new HashMapStateBackend());
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000);
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop101:8020/eduRealTime/ck");
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        System.setProperty("HADOOP_USER_NAME","atguigu");*/


        DataStreamSource<String> dataStreamSource = env.addSource(KafkaUtils.getFlinkKafkaConsumer("dwd_traffic_page_log", "dwd_traffic_page_log"));

        KeyedStream<JSONObject, String> keyedStream = dataStreamSource.map(JSON::parseObject)
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<JSONObject>forBoundedOutOfOrderness(Duration.ofMillis(200))
                        .withTimestampAssigner((e, r) -> e.getLong("ts")))
                .keyBy(value -> value.getJSONObject("common").getString("mid"));

        Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("begin")
                .where(new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        //先将last_page_id字段拿出来
                        String lastPageId = value.getJSONObject("page").getString("last_page_id");
                        //如果该字段是空的,说明这一条数据是整个会话的起点,留下
                        if (StringUtils.isEmpty(lastPageId)) {
                            return true;
                        }
                        return false;
                    }
                })
                .next("second")
                .where(new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        //先将last_page_id字段拿出来
                        String lastPageId = value.getJSONObject("page").getString("last_page_id");
                        //如果该字段是空的,说明这一条数据是整个会话的起点,留下
                        if (StringUtils.isEmpty(lastPageId)) {
                            return true;
                        }
                        return false;
                    }
                })
                //最后设置超时时间(详情见第三条判断方式)
                .within(org.apache.flink.streaming.api.windowing.time.Time.seconds(10));

        //超时数据将会被写入侧输出流中
        OutputTag<JSONObject> outputTag = new OutputTag<JSONObject>("lateData") {
        };
        PatternStream<JSONObject> patternStream = CEP.pattern(keyedStream, pattern);

        //根据自定义的函数类MyprocessFunction实现TimedOutPartialMatchHandler接口,可以处理迟到数据,将迟到数据写入侧输出流中
        SingleOutputStreamOperator<JSONObject> streamOperator = patternStream.process(new MyprocessFunction(outputTag));

        //从主流中获取侧输出流的数据
        DataStream<JSONObject> sideOutput = streamOperator.getSideOutput(outputTag);
        //将主流和侧输出流进行合并
        DataStream<JSONObject> union = streamOperator.union(sideOutput);
        //合并后对数据的格式进行转换,以便写入kafka  jsonStr --> jsonObj
        SingleOutputStreamOperator<String> stringSingleOutputStreamOperator = union.map(JSON::toString);


        //stringSingleOutputStreamOperator.print(">>>");
        stringSingleOutputStreamOperator.addSink(KafkaUtils.getFlinkKafkaProducer("dwd_traffic_user_jump_detail"));


        env.execute();

    }

    private static class MyprocessFunction extends PatternProcessFunction<JSONObject, JSONObject> implements TimedOutPartialMatchHandler<JSONObject> {
        //从主程序中将需要写入的侧输出流标签传入
        private OutputTag<JSONObject> outputTag ;

        public MyprocessFunction(OutputTag<JSONObject> outputTag) {
            this.outputTag = outputTag;
        }

        @Override
        public void processMatch(Map<String, List<JSONObject>> match, Context ctx, Collector<JSONObject> out) throws Exception {
            //将所有符合条件的数据过滤出来之后,将CEP中的第一条数据发送到下游(因为分析之后我们可以得知第一条数据是符合跳出操作的)
            JSONObject jsonObject = match.get("begin").get(0);
            out.collect(jsonObject);
        }

        @Override
        public void processTimedOutMatch(Map<String, List<JSONObject>> match, Context ctx) throws Exception {
            //处理迟到数据,将迟到数据写入侧输出流
            JSONObject begin = match.get("begin").get(0);
            ctx.output(outputTag,begin);
        }
    }

}
